from kafka import  KafkaConsumer
import time
import json
import queue
from config.config import KAFKA_HOST,PRE_DATA_TOPIC,AI_DATA_TOPIC,ALL_DATA_TOPIC

pre_data_queue = queue.Queue(maxsize=2000)
all_data_queue = queue.Queue(maxsize=2000)
ai_data_queue = queue.Queue(maxsize=2000)


# 接收自动驾驶车辆的实时数据
def receive_message_of_pre_data():
    print("接收自动驾驶车辆的实时数据", 'kafka 消费被触发')
    global pre_data_queue
    # 首先定义全局变量，用于在不同方法内互相通信
    consumer = KafkaConsumer(
        PRE_DATA_TOPIC,
        bootstrap_servers=KAFKA_HOST,
        auto_offset_reset='latest',
        enable_auto_commit=False
    )
    # 将消费者的偏移量设置为最后一条消息的偏移量
    """
            同步发送 数据
            :param data_li:  发送数据
            :return:
            """
    while True:
        time.sleep(0.1 / 6)
        for message in consumer:
            if message:
                try:
                    # message_key = message.key.decode('utf-8')
                    message_value = message.value.decode('utf-8')
                    vehicle_info = json.loads(message_value)
                    # print(type(vehicle_info),vehicle_info)
                    # print("kafka消费者接收到的真实车辆数据", vehicle_info)
                    if pre_data_queue.qsize() < 2000:
                        pre_data_queue.put(vehicle_info)
                except Exception as e:
                    print(str(e))


# 接收自动驾驶车辆的实时数据
def receive_message_of_all_data():
    print("接收all-data", 'kafka 消费被触发')
    global all_data_queue
    # 首先定义全局变量，用于在不同方法内互相通信
    consumer = KafkaConsumer(
        ALL_DATA_TOPIC,
        bootstrap_servers=KAFKA_HOST,
        auto_offset_reset='latest',
        enable_auto_commit=False
    )
    # 将消费者的偏移量设置为最后一条消息的偏移量
    """
            同步发送 数据
            :param data_li:  发送数据
            :return:
            """
    while True:
        time.sleep(0.1 / 6)
        for message in consumer:
            if message:
                try:
                    # message_key = message.key.decode('utf-8')
                    message_value = message.value.decode('utf-8')
                    vehicle_info = json.loads(message_value)
                    # print(type(vehicle_info),vehicle_info)
                    # print("kafka消费者接收到的真实车辆数据", vehicle_info)
                    if all_data_queue.qsize() < 2000:
                        all_data_queue.put(vehicle_info)
                except Exception as e:
                    print(str(e))
# receive_message_of_pre_data()

# 接收自动驾驶车辆的实时数据
def receive_message_of_ai_data():
    print("接收ai-data", 'kafka 消费被触发')
    global ai_data_queue
    # 首先定义全局变量，用于在不同方法内互相通信
    consumer = KafkaConsumer(
        AI_DATA_TOPIC,
        bootstrap_servers=KAFKA_HOST,
        auto_offset_reset='latest',
        enable_auto_commit=False
    )
    # 将消费者的偏移量设置为最后一条消息的偏移量
    """
            同步发送 数据
            :param data_li:  发送数据
            :return:
            """
    while True:
        time.sleep(0.1 / 6)
        for message in consumer:
            if message:
                try:
                    # message_key = message.key.decode('utf-8')
                    message_value = message.value.decode('utf-8')
                    vehicle_info = json.loads(message_value)
                    # print(type(vehicle_info),vehicle_info)
                    # print("kafka消费者接收到的真实车辆数据", vehicle_info)
                    if ai_data_queue.qsize() < 2000:
                        ai_data_queue.put(vehicle_info)
                except Exception as e:
                    print(str(e))